package org.onion.pattern.event.springboot.rocketmq.consume.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.onion.pattern.event.consumer.EventConsumeHandler;
import org.onion.pattern.event.exception.EventConsumeException;
import org.onion.pattern.event.spring.io.EventCodecs;
import org.onion.pattern.event.springboot.rocketmq.consume.TopicAndEventTypeMapping;

import java.util.List;

@Slf4j
public class RocketMessageOrderlyListener extends _MessageListener implements MessageListenerOrderly {

    public final static long SUSPEND_CURRENT_QUEUE_TIME_MILLIS = 30000;

    public RocketMessageOrderlyListener(DefaultMQPushConsumer consumer, EventConsumeHandler consumeHandler
            , EventCodecs eventCodecs, TopicAndEventTypeMapping topicAndEventTypeMapping) {
        this.consumeHandler = consumeHandler;
        this.consumer = consumer;
        this.eventCodecs = eventCodecs;
        this.topicAndEventTypeMapping = topicAndEventTypeMapping;
    }


    //ConsumeMessageOrderlyService
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        context.setSuspendCurrentQueueTimeMillis(SUSPEND_CURRENT_QUEUE_TIME_MILLIS);

        for (MessageExt messageExt : msgs) {
            log.debug("【事件消费】接收到事件, group->{}, topic->{}, tags->{}, message-id->{}, message-key->{}, queue->{}",
                    consumer.getConsumerGroup(), messageExt.getTopic(), messageExt.getTags(),
                    messageExt.getMsgId(),
                    messageExt.getKeys(), messageExt.getQueueId());

            int reconsumeTimes = messageExt.getReconsumeTimes() + 1;
            try {
                handleMessage(messageExt);
//                if (subscribeResult.getExecutionStatus() == ExecutionStatus.FAILURE
//                        || subscribeResult.getExecutionStatus() == ExecutionStatus.UNKNOWN) {
//                    String pattern = "【事件消费】事件消费失败[{0}]次, group->{1}, topic->{2}, tag->{3}, message_key->{4}, error->{5}";
//                    Object[] args = {failureCount, consumer.getConsumerGroup(), messageExt.getTopic(),
//                            messageExt.getTags(), messageExt.getKeys(),
//                            subscribeResult.getException() != null ? subscribeResult.getException().getMessage()
//                                    : "unknown"};
//                    if (subscribeResult.getException() instanceof DomainEventException) {
//                        log.warn(MessageFormat.format(pattern, args));
//                    } else {
//                        log.warn(MessageFormat.format(pattern, args), subscribeResult.getException());
//                    }
//
//                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
//                }
            } catch (EventConsumeException ex) {
                log.warn("【事件消费】失败[{}]次, group->{}, topic->{}, tag->{}, message_key->{}", reconsumeTimes, consumer.getConsumerGroup(), messageExt.getTopic(),
                        messageExt.getTags(), messageExt.getKeys(), ex);
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            } catch (Throwable ex) {
                log.error("【事件消费】失败[{}]次, group->{}, topic->{}, tag->{}, message_key->{}",
                        reconsumeTimes < 99 ? String.valueOf(reconsumeTimes) : "99+", consumer.getConsumerGroup(), messageExt.getTopic(),
                        messageExt.getTags(), messageExt.getKeys()
                        , ex);
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }

        }
        return ConsumeOrderlyStatus.SUCCESS;
    }


}
